package com.ds.infrastructure.hyperspace.container.endpoint.connector.health;

import com.ds.infrastructure.hyperspace.container.endpoint.destination.TargetEndpointNode;
import com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2Codec;
import com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2FrameTypeEnum;
import com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2FrameWrapper;
import com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2GoAwayFrame;
import io.netty.channel.*;
import io.netty.handler.codec.http2.DefaultHttp2GoAwayFrame;
import io.netty.handler.codec.http2.DefaultHttp2PingFrame;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import static com.ds.infrastructure.hyperspace.container.internal.http2.HyperspaceHttp2FrameTypeEnum.*;
import static io.netty.channel.ChannelFutureListener.CLOSE;

/**
 * @author  Weishaoying
 */
@Slf4j
public class Http2HealthCheckHandler extends ChannelDuplexHandler {

    private final HyperspaceHttp2FrameWrapper PING_FRAME_WRAPPER = new HyperspaceHttp2FrameWrapper(
            PING, new DefaultHttp2PingFrame(0L, false));

    private final HyperspaceHttp2FrameWrapper GO_AWAY_FRAME_WRAPPER = new HyperspaceHttp2FrameWrapper(
            GO_AWAY, new HyperspaceHttp2GoAwayFrame(0, new DefaultHttp2GoAwayFrame(0)));

    private final HyperspaceHttp2Codec hyperspaceHttp2Codec;

    private final TargetEndpointNode targetEndpointNode;

    private final HyperspaceNodeHealthChecker nodeHealthChecker;

    private final ChannelFutureListener pingChannelFutureListener;

    public Http2HealthCheckHandler(HyperspaceHttp2Codec hyperspaceHttp2Codec,
            TargetEndpointNode targetEndpointNode, HyperspaceNodeHealthChecker nodeHealthChecker) {
        this.hyperspaceHttp2Codec = hyperspaceHttp2Codec;
        this.targetEndpointNode = targetEndpointNode;
        this.nodeHealthChecker = nodeHealthChecker;
        this.pingChannelFutureListener = new PingChannelFutureListener();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        HyperspaceHttp2FrameWrapper http2FrameWrapper = (HyperspaceHttp2FrameWrapper) msg;
        HyperspaceHttp2FrameTypeEnum frameType = http2FrameWrapper.getFrameType();
        Object frameData = http2FrameWrapper.getFrameData();
        try {
            if (SETTINGS == frameType) {
                hyperspaceHttp2Codec.writeAndFlushData(ctx,
                        new HyperspaceHttp2FrameWrapper(SETTINGS, frameData));
                hyperspaceHttp2Codec.writeAndFlushSettingsAck(ctx);
                hyperspaceHttp2Codec.writeAndFlushData(ctx, PING_FRAME_WRAPPER)
                        .addListener(pingChannelFutureListener);
            } else if (PING_ACK == frameType) {
                targetEndpointNode.incrementPassCount(ctx.channel().remoteAddress());
                hyperspaceHttp2Codec.writeAndFlushData(ctx, GO_AWAY_FRAME_WRAPPER)
                        .addListener(CLOSE);
            }
        } finally {
            ReferenceCountUtil.release(frameData);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("Http2HealthCheckHandler failed, target node info({}),remote address:{}",
                targetEndpointNode.getKey(),ctx.channel().remoteAddress(), cause);
        targetEndpointNode.incrementFailCount(ctx.channel().remoteAddress());
        Channel channel = ctx.channel();
        if (channel.isOpen()) {
            channel.flush().close();
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelInactive();
        nodeHealthChecker.checkComplete();
    }

    class PingChannelFutureListener implements ChannelFutureListener {

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
                log.error("Ping target failed!target node info({}),remote address:{}",
                        targetEndpointNode.getKey(),future.channel().remoteAddress(), future.cause());
                targetEndpointNode.incrementFailCount(future.channel().remoteAddress());
            }
        }
    }

}
